package com.hll.iot;

import cn.hutool.core.thread.ThreadUtil;
import com.aos.iot.protocol.hll.message.HllMessage;
import com.aos.iot.socket.common.message.Message;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMapCache;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class HllMessageSendHelper {

    private final RedissonClient redissonClient;

    private final Map<String, RTopic> topicMap = new ConcurrentHashMap<>();

    private final Map<String, RMapCache<String, HllMessage>> returnMap = new ConcurrentHashMap<>();

    public HllMessageSendHelper(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }

    public Future<HllMessage> sendCallback(String clientId, HllMessage message) {
        if(!returnMap.containsKey(clientId)) {
            returnMap.put(clientId, redissonClient.getMapCache("IOT:RECEIVE:" + clientId));
        }
        if(!topicMap.containsKey(clientId)) {
            topicMap.put(clientId, redissonClient.getTopic(clientId));
        }
        message.setDirection(Message.Direction.DOWN);
        topicMap.get(clientId).publish(message);
        String msgId = message.getMsgId();
        RMapCache<String, HllMessage> receiveMap = returnMap.get(clientId);
        return ThreadUtil.execAsync(() -> {
            while (true) {
                ThreadUtil.sleep(100, TimeUnit.MILLISECONDS);
                if(receiveMap.containsKey(msgId)) {
                    return receiveMap.remove(msgId);
                }
            }
        });
    }

    public HllMessage sendAndReceive(String clientId, HllMessage message) {
        return sendAndReceive(clientId, message, 5);
    }

    public HllMessage sendAndReceive(String clientId, HllMessage message, int timeout) {
        Future<HllMessage> future = sendCallback(clientId, message);
        try {
            return future.get(timeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            return null;
        }
    }

    public void receive(String clientId, HllMessage message, int ttl) {
        returnMap.get(clientId).put(message.getMsgId(), message, ttl, TimeUnit.SECONDS);
    }

}
